package com.zyx.flinkdemo.dataset.iteration;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.IterativeDataSet;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/30 13:39
 * desc:
 */
public class IncrementingNumbers {
    public static void main(String[] args) throws Exception {
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        int maxIterations = 10;
        // Create initialIterativeDataSet
        DataSet<Long> input = env.generateSequence(1, 5);

        // 启动迭代
        IterativeDataSet<Long> initial = input.iterate(maxIterations);

        // 指定step fun
        DataSet<Long> iteration = initial.map(new MapFunction<Long, Long>() {
            @Override
            public Long map(Long i) throws Exception {
                return i + 1;
            }
        });

        // Iteratively transform the IterativeDataSet
        DataSet<Long> result = initial.closeWith(iteration);
        // 结果保存
        result.print();

        // execute program
        // env.execute("IterativePi Example");
    }
}
